package cn.doitedu.common.utils;

import cn.doitedu.demo7.RuleCalculator;
import cn.doitedu.demo7.RuleMeta;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Slf4j
public class CommonUtils {

    public static boolean compare(String conditionValue, String realValue, String compareType) {

        boolean res;
        switch (compareType) {
            case ">":
                res = realValue.compareTo(conditionValue) > 0;
                break;
            case ">=":
                res = realValue.compareTo(conditionValue) >= 0;
                break;
            case "<":
                res = realValue.compareTo(conditionValue) < 0;
                break;
            case "<=":
                res = realValue.compareTo(conditionValue) <= 0;
                break;
            case "<>":
                res = !realValue.equals(conditionValue);
                break;
            case "=":
                res = realValue.equals(conditionValue);
                break;
            case "contains":
                res = realValue.contains(conditionValue);
                break;
            case "regexp_match":
                Pattern pattern = Pattern.compile(conditionValue);
                Matcher matcher = pattern.matcher(realValue);
                res = matcher.matches();
                break;
            default:
                throw new RuntimeException("你有病吧，什么玩意我不认识!!!");

        }
        return res;
    }

    public static DataStreamSource<String> getActionJsonStream(StreamExecutionEnvironment env) {

        KafkaSource<String> source = KafkaSource
                .<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("user-action-log")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doitedu-ww")
                .setClientIdPrefix("doitedu-cc")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "log-source");


        return stream;
    }


    public static SingleOutputStreamOperator<RuleMeta> getRuleMetaStream(StreamExecutionEnvironment env) {
        MySqlSource<String> ruleMetaCdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit46") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit46.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        // 解析cdc抓取的规则元数据变更
        DataStreamSource<String> cdcStream = env.fromSource(ruleMetaCdcSource, WatermarkStrategy.noWatermarks(), "cdc");

        // 对抓取到的ruleMeta变更原始json数据，进行解析、提取
        SingleOutputStreamOperator<RuleMeta> ruleMetaBeans = cdcStream.map(json -> {

            JSONObject jsonObject = JSON.parseObject(json);
            String op = jsonObject.getString("op");

            JSONObject dataObj;
            if (op.equals("d")) {
                dataObj = jsonObject.getJSONObject("before");
            } else {
                dataObj = jsonObject.getJSONObject("after");
            }

            byte[] bytes = dataObj.getBytes("pre_select_users");

            String jsonString = dataObj.toJSONString();
            RuleMeta ruleMeta = JSON.parseObject(jsonString, RuleMeta.class);

            // 填充op字段
            ruleMeta.setOp(op);

            // 反序列化 bitmap的字节数组
            byte[] preSelectUsersBytes = ruleMeta.getPre_select_users();
            Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();

            if (preSelectUsersBytes != null) {
                bitmap.deserialize(ByteBuffer.wrap(preSelectUsersBytes));
                log.info("反序列化得到的bitmap是: {}", bitmap);
            }
            // 填充bitmap字段
            ruleMeta.setPreSelectBitmap(bitmap);

            return ruleMeta;
        });


        return ruleMetaBeans;
    }


    public static void restoreCalculatorPool(GroovyClassLoader groovyClassLoader,
                                      HashMap<String, RuleCalculator> ruleCalculatorPool,
                                      ReadOnlyBroadcastState<String, RuleMeta> broadcastState,
                                      RuntimeContext runtimeContext) throws Exception {

        if (ruleCalculatorPool.isEmpty()) {
            for (Map.Entry<String, RuleMeta> entry : broadcastState.immutableEntries()) {
                // 遍历广播状态中的每一条规则元数据，来构造、初始化运算机对象
                String ruleId = entry.getKey();
                RuleMeta meta = entry.getValue();

                String modelCode = meta.getModel_code();
                Class<?> aClass = groovyClassLoader.parseClass(modelCode);
                RuleCalculator calculator = (RuleCalculator) aClass.newInstance();
                calculator.init(runtimeContext, meta);

                // 将初始化好的运算机对象，放入运算机池
                ruleCalculatorPool.put(ruleId, calculator);

                log.warn("恢复了一个规则运算机对象,rule_id:{}",ruleId);

            }
        }
    }


}
